//
// Created by hjie on 23-5-7.
//
#include "redis_client.h"
#include "redis_message_reply.h"
#include "redis_message.h"
#include <cstdio>
#include <functional>
#include <mutex>
#include <unistd.h>

RedisClient::RedisClient(LoopEvent * loop, const std::string &ip, uint16_t port)
{
    m_loop_event = loop;

    m_option.m_port = port;
    m_option.m_server_ip = ip;

    m_current_iter = m_response_map.end();

    ConnectToRedis();
}

RedisClient::RedisClient(LoopEvent *loop, RedisOption &option)
{
    m_option = option;
    m_current_iter = m_response_map.end();
    ConnectToRedis();
}

RedisClient::~RedisClient()
{
    if (m_connection_client)
    {
        delete m_connection_client;
        m_connection_client = nullptr;
    }
    m_response_map.clear();
}

bool RedisClient::ConnectToRedis()
{
    m_connection_client =  new Client(m_loop_event);
    m_connection_client->SetOnConnection(std::bind(&RedisClient::OnConnection, this));
    m_connection_client->SetDisConnection(std::bind(&RedisClient::OnDisconnection, this, std::placeholders::_1));
    m_connection_client->SetOnMessage(std::bind(&RedisClient::OnMessage, this, std::placeholders::_1, std::placeholders::_2));
    m_connection_client->Connect(m_option.m_server_ip, m_option.m_port);
    m_client_status = false;
    return true;
}

bool RedisClient::CheckClientStatus()
{
    return m_client_status.load();
}

int64_t RedisClient::OperateCommand(const std::string &msg)
{
    if (!CheckClientStatus())
    {
        // reconnection to server
        return -2;
    }
    RedisMessageResponse * response = new RedisMessageResponse();
    bool result = response->m_request.ParseRequest(msg);
    if (!result)
    {
        // parse parameter failing
        delete response;
        response = nullptr;
        return -1;
    }
    std::string content = response->m_request.GetData();
    if (!content.empty())
    {
        std::unique_lock<std::mutex> lk(m_mtx);   
        m_connection_client->SendMessage(content);
        int64_t index = m_index++;
        
        m_response_map[index]          = response;
        m_response_map[index]->m_index = index;
        if (m_current_iter == m_response_map.end())
        {
            m_current_iter = m_response_map.find(index);
        }
        return index;
    }
    delete response;
    response = nullptr;
    return -1;
}

void RedisClient::OnConnection()
{
    m_client_status.store(true);
}

void RedisClient::RelaseAllResponse()
{
    {
        std::unique_lock<std::mutex> lk(m_mtx);
        for (auto iter = m_response_map.begin(); iter != m_response_map.end(); iter++)
        {
            iter->second->m_conf.notify_all();
        }
    }
    {
        std::unique_lock<std::mutex> lk(m_mtx);
        for (auto iter = m_response_map.begin(); iter != m_response_map.end(); )
        {
            auto use_iter = iter++;
            delete use_iter->second;
        }
        m_current_iter = m_response_map.end();
    }
}

void RedisClient::OnDisconnection(int code)
{
    m_client_status.store(false);
    m_loop_event->PushTask([this](){

        if (m_connection_client)
        {
            RelaseAllResponse();
            // process all response data
            delete m_connection_client;
            m_connection_client = nullptr;
            // auto reconnection to the redis server
            ConnectToRedis();
        }
    });
}

void RedisClient::OnMessage(const uint8_t * data, int length)
{
    // printf("recv data %s and length %d\n", data, length);
    // 需要重复解析
    do
    {
        int use_size = 0;
        bool result = DecoderMsg(data, length, use_size);
        if (result && use_size > 0)
        {
            m_connection_client->m_data->RemoveData(use_size);
            length -= use_size;
            data   += use_size;
            if (length < 3) 
            {
                break;
            }
            continue;
        }
        else if (use_size > 0)
        {
            m_connection_client->m_data->RemoveData(use_size);
        }
        else 
        {
            std::cout << __func__ << std::endl;
            break;
        }

    }while(1);
}

RedisMessageResponse *RedisClient::GetReplyByIndex(int64_t index)
{
    RedisMessageResponse * response = nullptr;
    std::unique_lock<std::mutex> lk(m_mtx);
    if (m_current_iter != m_response_map.end() && m_current_iter->second && m_current_iter->second->m_index == index)
    {
        response = m_current_iter->second;
        m_current_iter = m_response_map.begin();
    }
    else
    {
        auto iter_data = m_response_map.find(index);
        if (iter_data == m_response_map.end() || iter_data->second == nullptr)
        {
            return nullptr;
        }
        response = iter_data->second;
    }
    lk.unlock();

    std::unique_lock<std::mutex> self_lk(response->m_mtx);
    response->m_conf.wait(self_lk, [response, this](){

        if (response->m_reply == nullptr && CheckClientStatus())
        {
            return false;
        }
        return true;

    });
    return response;
}

void RedisClient::FreeReplyByIndex(int64_t index)
{
    std::unique_lock<std::mutex> lk(m_mtx);
    auto iter = m_response_map.find(index);
    if (iter != m_response_map.end())
    {
        delete iter->second;
        m_response_map.erase(iter);
    }
}

bool RedisClient::DecoderMsg(const uint8_t *data, int length, int &use_data_size)
{
    auto * reply = new RedisMessageReply();
    int use_size = 0;
    auto result  = reply->ParseData(data, length, use_size);
    switch (result)
    {
        case EmParseStatusNone:
        {
            // nothing to do
            break;
        }
        case EmParseStatusFinish:
        {
            use_data_size = use_size;
            std::unique_lock<std::mutex> lk(m_mtx);
            if (m_current_iter != m_response_map.end() && m_current_iter->second == nullptr)
            {
                m_current_iter->second->m_reply = reply;
                m_current_iter->second->m_conf.notify_one();
            }
            else
            {
                for (auto iter = m_response_map.begin(); iter != m_response_map.end(); iter++)
                {
                    if (iter->second->m_reply == nullptr && iter->second->m_is_async == 0)
                    {
                        iter->second->m_reply = reply;
                        iter->second->m_conf.notify_one();
                        break;
                    }
                }
            }
            break;
        }
        case EmParseStatusNeedData:
        {
            use_data_size = 0;
            break;
        }
        case EmParseStatusError:
        {
            // remove first char data
            use_data_size = 1;
            break;
        }
    }
    if (result != EmParseStatusFinish)
    {
        delete reply;
        reply = nullptr;
        return false;
    }
    return true;
}

